package com.intct.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.common.Constant;
import com.intct.utils.RowKeyUilt;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.HashMap;
import java.util.Map;

/**
 * @author gufg
 * @since 2025-09-22 11:20
 *
 * Rowkey设计：
 *      原则：唯一性、散列性、尽量简短
 *      常用设计方式：数据反转、加盐、字符串拼接
 *
 * 预分区设计：
 *
 * Rowkey按照字典序
 *
 * 项目中用到的就是字符串拼接(轨迹)：
 *    rowkey: order_id + time
 *            如：175927190436821475976916
 *    做8个分区
 *                    起始rowkey                结束rowkey
 *    第1分区          空值                      0001
 *    第2分区          0001                     0002
 *    第3分区          0002                     0003
 *    第4分区          0003                     0004
 *    第5分区          0004                     0005
 *    第6分区          0005                     0006
 *    第7分区          0006                     0007
 *    第8分区          0007                     空值
 * 000734234234
 * 12332322323
 * 99999999999
 * hbashcode(order_id) % 8 ==> 5(左补零) ==> 0005 + substr(md5(17592719043682), 1, 12) == 000532423423423
 */
public class AdsLocusMapFunc extends RichMapFunction<String, String> {
    private Map<String, String> configMap = new HashMap<>();

    @Override
    public void open(Configuration parameters) throws Exception {
        // JDBC -- read all travel_config.dim_table_config
        Class.forName("com.mysql.cj.jdbc.Driver");
        Connection jdbcConn = DriverManager.getConnection("jdbc:mysql://cdh-node:13306/travel_config", "root", "Test_090110");

        PreparedStatement preparedStatement = jdbcConn.prepareStatement("select * from travel_config.ads_table_config");
        ResultSet resultSet = preparedStatement.executeQuery();

        while (resultSet.next()) {
            String key = resultSet.getString("source_table");
            JSONObject jsonObj = new JSONObject();

            JSONObject dataObj = new JSONObject();
            dataObj.put("source_table", resultSet.getString("source_table"));
            dataObj.put("sink_table", resultSet.getString("sink_table"));
            dataObj.put("sink_family", resultSet.getString("sink_family"));
            dataObj.put("sink_columns", resultSet.getString("sink_columns"));
            dataObj.put("sink_row_key", resultSet.getString("sink_row_key"));
            jsonObj.put(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY, dataObj);
            configMap.put(key, jsonObj.toJSONString());
        }

        resultSet.close();
        preparedStatement.close();
        jdbcConn.close();
    }

    @Override
    public String map(String jsonStr) throws Exception {
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        // {"driver_id":"11","lng":"104.05166","create_time":"2019-04-19 13:55:37","driver_type":"1","mobile":"1861***1024","driver_name":"邵5bfb8fd1c7af69efee217c72fabec38b","cancel_count":"0","oder_id_desen":"175********682","register_city":"110108","time":"1475976832","id":"11","order_id":"17592719043682","lat":"30.70214","driver_management_id":"0"}

        JSONObject newObj = new JSONObject(true);

        // 插入数据
        newObj.put(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY, jsonObj);
        // 配置表数据
        // // {"rowkey":"511603","sink_family":"info","after":{"code":"511603","name":"四川省广安市前锋区"},"table":"dim_area_info"}

        String locus = configMap.get("locus");
        JSONObject mapAfterJson = JSON.parseObject(locus);
        JSONObject data = mapAfterJson.getJSONObject(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY);

        newObj.put("table",data.get("sink_table"));
        newObj.put("sink_family",data.get("sink_family"));

        // 多个字段做为Rowkey时，需要特殊处理
        String[] rowkeyColumnList = data.getString("sink_row_key").split(",");
        String rowkey = "";
        for (String rowkeyColumn : rowkeyColumnList) {
            String value = jsonObj.getString(rowkeyColumn);
            rowkey = value + ":" + rowkey;
        }
        newObj.put("rowkey", RowKeyUilt.getRowkey(rowkey, 8));

        return newObj.toJSONString();
    }
}
